package com.deep.flink;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author sw
 * @create 2023-06-30 11:26
 */
public class Test4 {
    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);

        // 完全不依赖env
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();

        TableEnvironment tableEnv = TableEnvironment.create(settings);

        //2.创建表
        String createDDL = "CREATE TABLE clickTable (" +
                " user_name STRING," +
                " url STRING," +
                " ts BIGINT " +
                ") WITH (" +
                " 'connector' = 'filesystem'," +
                " 'path' = 'input/clicks.txt" +
                " 'format' = 'csv'" +
                ")";

        tableEnv.executeSql(createDDL);

        //创建用于输出的表
        String createOutDDL = "CREATE TABLE outTable (" +
                " url STRING," +
                " user_name STRING " +
                ") WITH (" +
                " 'connector' = 'filesystem'," +
                " 'path' = 'output'," +
                " 'format' = 'csv'" +
                ")";
        TableResult tableResult = tableEnv.executeSql(createOutDDL);

        tableEnv.useCatalog("custom_catalog");
        tableEnv.useDatabase("custom_database");

        Table clickTable = tableEnv.from("clickTable")
//                .where($("user_name").isEqual("zhangsan"))
                .select($("user_name"), $("url"));

        TableResult outTable = clickTable.executeInsert("outTable");

        tableEnv.execute("test");
    }
}
